package pbc.gx.SaveJson.kafka.producer;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.JSONReader;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.*;

import java.io.FileReader;
import java.text.DecimalFormat;
import java.util.ArrayList;
import java.util.List;

@RestController
@CrossOrigin
@Slf4j
public class KafkaProducer {
    private final static String TOPIC_NAME = "jsonTopic";

    private final KafkaTemplate<String, JSONObject> kafkaTemplate;
    private final RedisTemplate redisTemplate;


    public KafkaProducer(KafkaTemplate<String, JSONObject> kafkaTemplate, RedisTemplate redisTemplate) {
        this.kafkaTemplate = kafkaTemplate;
        this.redisTemplate = redisTemplate;
    }

    private void send2Kafka(List<JSONObject> jsons, String tableName) {
        int size = jsons.size();
        Long maxId = redisTemplate.opsForValue().increment("id:" + tableName, jsons.size());
        DecimalFormat df = new DecimalFormat("00000000000000000000");
        for (JSONObject json : jsons) {
            Long id = (maxId - size);
            json.put("id", id);
            //log.info(df.format(id));
            kafkaTemplate.send(TOPIC_NAME, tableName + ":" + df.format(id), json);
            size--;
        }
    }

    @RequestMapping("/send")
    public String send(@RequestParam(name = "count", required = false) Integer count) {
        //发送功能就一行代码~
        FileReader fileReader = null;
        JSONReader jsonReader = null;
        if (count == null || count <= 0) {
            count = 1;
        }
        try {
            int n = 0;
            for (int i = 0; i < count; i++) {
                String filePath = "C:/Users/Administrator/Desktop/dev.json";
                fileReader = new FileReader(filePath);
                jsonReader = new JSONReader(fileReader);
                jsonReader.startArray();
                String fileName = filePath.substring(filePath.lastIndexOf("/") + 1);
                String tableName = fileName.substring(0, fileName.indexOf("."));
                List<JSONObject> jsons = new ArrayList<>();
                while (jsonReader.hasNext()) {
                    jsons.add((JSONObject) jsonReader.readObject());
                    if (jsons.size() >= 1000) {
                        send2Kafka(jsons, tableName);
                        n += jsons.size();
                        jsons.clear();
                        jsons = new ArrayList<>();
                    }
                }
                send2Kafka(jsons, tableName);
                n += jsons.size();
                jsonReader.endArray();
                jsonReader.close();
                fileReader.close();
            }
            return "send:" + n + " records";
        } catch (Exception e) {
            e.printStackTrace();
            return e.getMessage();
        }
    }
}
